/**
 * Copyright (C), 2015-2018, XXX有限公司
 * FileName: StormKafkaTopo
 * Author:   An-Il
 * Date:     2018/11/16 22:41
 * Description: storm整合kafka
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.blog.storm.example.integration.kafka.consumer;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;

/**
 * 〈一句话功能简述〉<br>
 * 〈storm整合kafka 0.10.x+〉
 *
 * @author An-Il
 * @create 2018/11/16
 * @url http://storm.apache.org/releases/1.2.2/storm-kafka-client.html
 * @since 1.0.0
 */
public class StormKafkaTopo {

    public static void main(String[] args) {

        final TopologyBuilder topologyBuilder = new TopologyBuilder();

        //包含“topic”，“partition”，“offset”，“key”，“value”字段。
        String host = "192.168.31.140:9092";
        String topic = "storm_kafka_test";

        KafkaSpoutConfig<String, String> kafkaSpoutConfig = KafkaSpoutConfig.builder(host, topic).build();

        topologyBuilder.setSpout("kafka_spout", new KafkaSpout<>(kafkaSpoutConfig));
        topologyBuilder.setBolt("my_bolt", new myBolt()).shuffleGrouping("kafka_spout");

        //本地测试
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("LocalWordCountJdbcApp", new Config(), topologyBuilder.createTopology());
    }
}